package day08;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import po.Event;
import po.UserEvent;

import java.util.List;
import java.util.Map;

/**
 * @Description:  使用flink-cep检测连续三次登录失败
 * @Author: ZYX
 * @Date: 2022/2/18 15:28
 * @Version: 1.0
 */
public class Demo03 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<UserEvent> stream = env.fromElements(new UserEvent("user-1", "fail", 1000L),
                new UserEvent("user-1", "fail", 2000L),
                new UserEvent("user-2", "success", 3000L),
                new UserEvent("user-1", "fail", 4000L))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<UserEvent>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<UserEvent>() {
                                    @Override
                                    public long extractTimestamp(UserEvent userEvent, long l) {
                                        return userEvent.timestamp;
                                    }
                                })
                );
        // 定义模板
        Pattern<UserEvent, UserEvent> pattern = Pattern.<UserEvent>begin("first")
                .where(new SimpleCondition<UserEvent>() {
                    @Override
                    public boolean filter(UserEvent userEvent) throws Exception {
                        return userEvent.eventType.equalsIgnoreCase("fail");
                    }
                })
                // next 严格紧邻
                .next("second")
                .where(new SimpleCondition<UserEvent>() {
                    @Override
                    public boolean filter(UserEvent userEvent) throws Exception {
                        return userEvent.eventType.equalsIgnoreCase("fail");
                    }
                }).next("third")
                .where(new SimpleCondition<UserEvent>() {
                    @Override
                    public boolean filter(UserEvent userEvent) throws Exception {
                        return userEvent.eventType.equalsIgnoreCase("fail");
                    }
                });

        // 在流上匹配模板
        PatternStream<UserEvent> patternStream = CEP.pattern(stream.keyBy(obj -> obj.userId), pattern);

        // 使用select方法将匹配事件取出
        patternStream.select(new PatternSelectFunction<UserEvent, String>() {
            @Override
            public String select(Map<String, List<UserEvent>> map) throws Exception {
                // Map的key是给事件起的名字
                // 列表是名字对应的事件所构成的列表
                UserEvent first = map.get("first").get(0);
                UserEvent second = map.get("second").get(0);
                UserEvent third = map.get("third").get(0);
                String result = "用户："+first.userId+" 在时间段： "+first.timestamp+";"+second.timestamp+";"+third.timestamp+";"
                        +"登录失败了！";
                return result;
            }
        }).print();


        env.execute();

    }

}
